package mylibrary.io.nio.nonBlock.RoxSample;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

public class NioServer implements Runnable{

	private InetAddress hostAddress;
	private int port;
	
	private ServerSocketChannel serverChannel;
	
	private Selector selector;
	
	private ByteBuffer readBuffer = ByteBuffer.allocate(8192);
	
	private EchoWorker worker;
	
	private List changeRequests = new LinkedList();
	
	private Map pendingData = new HashMap();
	
	public NioServer(InetAddress hostAddress, int port, EchoWorker worker) throws Exception {
		this.hostAddress = hostAddress;
		this.port = port;
		this.selector = this.initSelector();
		this.worker = worker;
	}
	
	private Selector initSelector() throws Exception {
		Selector selector = SelectorProvider.provider().openSelector();
		
		this.serverChannel = ServerSocketChannel.open();
		serverChannel.configureBlocking(false);
		
		InetSocketAddress isa = new InetSocketAddress(this.hostAddress, this.port);
		serverChannel.socket().bind(isa);
		
		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
		
		return selector;
	}
	
	
	public void run() {
		while (true) {
			try {
				
				synchronized(this.changeRequests) {
					Iterator changes = this.changeRequests.iterator();
					while(changes.hasNext()) {
						ChangeRequest change = (ChangeRequest) changes.next();
						switch(change.type) {
							case ChangeRequest.CHANGEOPS:
							SelectionKey key = change.socket.keyFor(selector);
							key.interestOps(change.ops);
						}
					}
					this.changeRequests.clear();
				}
				
				this.selector.select();
				
				Iterator selectedKeys = this.selector.selectedKeys().iterator();
				
				while(selectedKeys.hasNext()) {
					SelectionKey key = (SelectionKey) selectedKeys.next();
					selectedKeys.remove();
					
					if(!key.isValid()) {
						continue;
					}
					
					if(key.isAcceptable()) {
						this.accept(key);
					} else if(key.isReadable()) {
						this.read(key);
					} else if(key.isWritable()) {
						this.write(key);
						
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private void accept(SelectionKey key) throws IOException {
		
		 // For an accept to be pending the channel must be a server socket channel.
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
		
		SocketChannel socketChannel = serverSocketChannel.accept();
		Socket socket = socketChannel.socket();
		socketChannel.configureBlocking(false);
		
		socketChannel.register(this.selector, SelectionKey.OP_READ);
	}
	
	private void read(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		
		this.readBuffer.clear();
		
		int numRead;
		
		try {
			numRead = socketChannel.read(readBuffer);
		} catch (Exception e) {
			key.cancel();
			socketChannel.close();
			e.printStackTrace();
			return;
		}
		if(numRead == -1) {
			key.channel().close();
			key.cancel();
			return;
		}
		
		this.worker.processData(this, socketChannel, readBuffer.array(), numRead);
	}
	
	public void write(SelectionKey key) throws IOException {
		
		SocketChannel channel = (SocketChannel) key.channel();
		synchronized(this.pendingData) {
			List queue = (List) this.pendingData.get(channel);
			
			while(!queue.isEmpty()) {
				ByteBuffer buff = (ByteBuffer) queue.get(0);
				channel.write(buff);
				
				if(buff.remaining()>0) {
					break;
				}
				queue.remove(0);
			}
			
			if(queue.isEmpty()) {
				key.interestOps(SelectionKey.OP_READ);
			}
		}
	}
	
	public void send(SocketChannel socket, byte[] data) {
		
		this.changeRequests.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
		
		List queue = (List) this.pendingData.get(socket);
		
		if(queue == null) {
			queue = new ArrayList();
			this.pendingData.put(socket, queue);
		}
		queue.add(ByteBuffer.wrap(data));
		
		this.selector.wakeup();
	}
	
	public static void main(String[] args) {
		try {
			
			EchoWorker worker = new EchoWorker();
			new Thread(worker).start();
			new Thread(new NioServer(null, 9090, worker)).start();	
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
}
